package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo14MaYi {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("作业社保演示")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import sparkSession.implicits._
    import org.apache.spark.sql.functions._

    //用户碳排放量表
    val ant_user_low_carbon: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", "\t")
      .schema("user_id STRING,data_dt STRING,low_carbon DOUBLE")
      .load("spark/data/ant_user_low_carbon.txt")

    val ant_plant_carbon: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", "\t")
      .schema("plant_id STRING,plant_name STRING,plant_carbon DOUBLE")
      .load("spark/data/ant_plant_carbon.txt")

    //        ant_user_low_carbon.show()
    //        ant_plant_carbon.show()

    /**
     * 蚂蚁森林植物申领统计
     * 问题：假设2017年1月1日开始记录低碳数据（user_low_carbon），假设2017年10月1日之前满足申领条件的用户都申领了一颗p004-胡杨，
     * 剩余的能量全部用来领取“p002-沙柳” 。
     * 统计在10月1日累计申领“p002-沙柳” 排名前10的用户信息；以及他比后一名多领了几颗沙柳。
     * 得到的统计结果如下表样式：
     */
    //因为用户能量表与植物能量表没有关联字段，如果想要在sql中使用胡杨以及沙柳的能量的话，需要先单独从植物能量表将胡杨以及沙柳的能量取出来，使用变量接收
    val huYangCarbon: Double = ant_plant_carbon
      //过滤条件是胡杨
      .where($"plant_name" === "胡杨")
      //选择能量的一列
      .select($"plant_carbon")
      //转rdd
      .rdd
      //转scala数组
      .collect()
      .map {
        case Row(plant_carbon: Double) => plant_carbon
      }.head

    val shaLiuCarbon: Double = ant_plant_carbon
      //过滤条件是胡杨
      .where($"plant_name" === "沙柳")
      //选择能量的一列
      .select($"plant_carbon")
      //转rdd
      .rdd
      //转scala数组
      .collect()
      .map {
        case Row(plant_carbon: Double) => plant_carbon
      }.head

    println(s"胡杨所需的碳排放量：${huYangCarbon},沙柳所需要的碳排放量：${shaLiuCarbon}")
    println("---------------------------------------------------------------------------")
    ant_user_low_carbon
      //取出2017年1月1日到2017年10月1日
      .where($"data_dt" >= "2017/1/1" and $"data_dt" <= "2017/10/1")
      //按照用户分组，求用户的总的排放量
      .groupBy($"user_id")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //判断能量是否满足一个胡杨的能量,如果满足直接减去胡杨的能量，得到剩余的能量
      .withColumn("shengYu_carbon", when($"sum_low_carbon" >= huYangCarbon, $"sum_low_carbon" - huYangCarbon).otherwise($"sum_low_carbon"))
      //计算剩余的能量可以领取多少个沙柳
      .withColumn("number", floor($"shengYu_carbon" / shaLiuCarbon))
      //.show()
      //获取后一名用户的领取沙柳的个数
      .withColumn("lead_number", lead($"number", 1, 0) over Window.orderBy($"number".desc))
      //.show()
      .withColumn("duo", $"number" - $"lead_number")
      .limit(10)
    //.show()

    /**
     * 蚂蚁森林低碳用户排名分析
     * 问题：查询user_low_carbon表中每日流水记录，条件为：
     * 用户在2017年，连续三天（或以上）的天数里，
     * 每天减少碳排放（low_carbon）都超过100g的用户低碳流水。
     * 需要查询返回满足以上条件的user_low_carbon表中的记录流水。
     * 例如用户u_002符合条件的记录如下，因为2017/1/2~2017/1/5连续四天的碳排放量之和都大于等于100g：
     */
    //("user_id STRING,data_dt STRING,low_carbon DOUBLE")
    ant_user_low_carbon
      //过滤出2017年的数据
      .where(substring($"data_dt", 0, 4) === "2017")
      //计算用户每天积攒的能量 用户id和日期一起分组
      .groupBy($"user_id", $"data_dt")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //过滤掉能量小于等于100的
      .where($"sum_low_carbon" > 100)
      //开窗有，以用户开窗，以日期升序排序，编号 row_number
      .withColumn("rn", row_number() over Window.partitionBy($"user_id").orderBy($"data_dt"))
      //使用日期减去排序
      .withColumn("flag_dt", date_sub(regexp_replace($"data_dt", "/", "-"), $"rn"))
      //计算连续的天数
      .withColumn("lianxu_days", count($"data_dt") over Window.partitionBy($"user_id", $"flag_dt"))
      //过滤出连续3天以上的数据
      .where($"lianxu_days" >= 3)
      //关联流水记录表，得到每个符合条件的数据
      .join(ant_user_low_carbon,List("user_id","data_dt"))
      .select($"user_id",$"data_dt",$"low_carbon")
      .show(100)


  }

}
